package com.qkjia.consumer;

import com.alibaba.fastjson.JSON;
import com.qkjia.constant.KafkaTopicConstant;
import com.qkjia.converter.ConsumerRecord2KafkaConsumerEvent;
import com.qkjia.converter.KafkaConsumerEvent;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
 * @author zhaoqiankun
 * @date 2021/11/10
 */
@Component
public class SpringConsumerTest {

    /**
     *  接受 kafka 消息
     *  @Payload String msg
     *  将方法参数绑定到消息有效负载的注释。 还可用于将有效负载与方法调用相关联。
     *  可以通过MessageConverter传递有效负载，以将其从具有特定 MIME 类型的序列化形式转换为与目标方法参数匹配的对象
     *  参数  接受到的消息
     *  例子： String ATTENDANCE_PUNCH_ADD = "attendance_punch_add";
     *  @KafkaListener(topicPattern = KafkaTopicConstant.ATTENDANCE_PUNCH_ADD)
     */
    @KafkaListener(topicPattern = KafkaTopicConstant.TEST_SEND)
    public void receive (@Payload String msg){
        System.out.println("接收到的消息" + msg);

    }
 /*   @KafkaListener(topicPattern = KafkaTopicConstant.TEST_SEND)
    public void receive (@Payload ConsumerRecord<Long, String> record){
        KafkaConsumerEvent event = ConsumerRecord2KafkaConsumerEvent.convert(record, this);
        System.out.println("receive:event{}" + event);
        Map map = JSON.parseObject((String) event.getMessage());
        map.get("").toString();

    }*/


}
